/**
 *Copyright [2010] [lcj(caijunlu@gmail.com)]
 *Licensed under the Apache License, Version 2.0 (the "License");
 *you may not use this file except in compliance with the License. 
 *You may obtain a copy of the License at 
 *             http://www.apache.org/licenses/LICENSE-2.0 
 *Unless required by applicable law or agreed to in writing, 
 *software distributed under the License is distributed on an "AS IS" BASIS, 
 *WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, 
 *either express or implied. See the License for the specific language governing permissions and limitations under the License
 */
package org.independent.servlet.session.util.kyro;

import java.nio.BufferOverflowException;
import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
import java.util.Map;

import org.independent.servlet.session.manager.SessionsManager;
import org.independent.servlet.session.service.ContextHolderFactory;
import org.independent.servlet.session.util.ObjectClasses;
import org.independent.servlet.session.util.XMemcachedClient;

import com.esotericsoftware.kryo.Kryo;
import com.esotericsoftware.kryo.serialize.ArraySerializer;
import com.esotericsoftware.kryo.serialize.MapSerializer;

/**
 * use Kryo to serialize and deserialize
 * 
 * @author lcj
 * 
 */
public class KryoUitl {
	private static XMemcachedClient xmc = SessionsManager.xMemcachedClient;

	/**
	 * all of the classes that need to serialize or deserialize are stored on
	 * the memcached, the key is dynamic,the key's key is sizeMark
	 */
	final static private String sizeMark = "kryo_size_mark";

	/**
	 * cache the Kryo on local
	 */
	public static Kryo kryo;

	/**
	 * the key for kryo be stored on memcache
	 */
	public static int kryoMarkSize;

	/**
	 * cache the Kryo registered classes
	 */
	public static ArrayList<Class<?>> kryoClasses = new ArrayList<Class<?>>();

	public static ArrayList<String> kryoClassesName = new ArrayList<String>();
   final static private void  registerInit(Kryo kryo){
	   ArraySerializer arraySerializer = new ArraySerializer(kryo);
		MapSerializer ms = new MapSerializer(kryo);
		ms.setKeysCanBeNull(false);
		ms.setValuesCanBeNull(false);
		kryo.register(HashMap.class, ms);
		kryo.register(int[].class, arraySerializer);
		kryo.register(byte[].class, arraySerializer);
		kryo.register(org.independent.servlet.session.ServletSession.class);
		kryo.register(org.independent.servlet.session.util.kyro.ByteArrayWarper.class);
		kryo.register(java.util.HashMap.class);
    }

	/**
	 * to register the classes of the value include to the Kryo this method be
	 * invoke by {@link org.independent.servlet.session.ServletSession} use
	 * optimistic locking to keep the Kryo on remote memcahed is right
	 * 
	 * @param value
	 *            the object be registered
	 */
	final static public void registerObjectClasses(Object value) {
		Class<?> clazz = value.getClass();
		/**
		 * if the class of object not in the cache,check it from memcached if the
		 * has been on the cache nothing to do
		 */
		if (!kryoClasses.contains(clazz)) {
			int size = 0;
			Object markSize;
			try {
				markSize = xmc.get(sizeMark);
				size = Integer.parseInt(markSize == null ? "0" : markSize + "");
			} catch (Exception e1) {
				e1.printStackTrace();
			}

			/**
			 * if the size be cached for local smaller than the size from the
			 * memcached then the local cache need update else if the size be
			 * cached for local is equal to the size from the memcached register
			 * the class use the kryo form memcached
			 */
			if (kryoMarkSize < size && size != 0) {
				try {
					updateLocalCache();
					if (kryoClasses.contains(clazz)) {
						commitKryoForNewClass(value);
					}
				} catch (NumberFormatException e) {
					e.printStackTrace();
				} catch (Exception e) {
					e.printStackTrace();
				}

			} else {
				commitKryoForNewClass(value);
			}
		}
	}
    static private ArrayList<String> baseClassesName=new ArrayList<String>();
    static{
    	baseClassesName.add("boolean");
    	baseClassesName.add("byte");
    	baseClassesName.add("char");
    	baseClassesName.add("short");
    	baseClassesName.add("int");
    	baseClassesName.add("long");
    	baseClassesName.add("float");
    	baseClassesName.add("double");
    	
    	baseClassesName.add("java.lang.Boolean");
    	baseClassesName.add("java.lang.Byte");
    	baseClassesName.add("java.lang.Character");
    	baseClassesName.add("java.lang.Short");
    	baseClassesName.add("java.lang.Integer");
    	baseClassesName.add("java.lang.Long");
    	baseClassesName.add("java.lang.Float");
    	baseClassesName.add("java.lang.Double");
    }
	/**
	 * Register the value include all classes to the kryo and update it to
	 * remote memcached
	 * 
	 * @param value
	 *            the object to register
	 * @param mark
	 *            whether kryo update success ; the value 1 is success ,the
	 *            value 0 is Failure
	 */
	final static private void commitKryoForNewClass(final Object value) {
		int mark = 0;
		while (mark == 0) {
			ContextHolderFactory.getServiceContextHolder().getContext()
					.clearMarkObjectClass();
			ArrayList<Class<?>> classesList = ObjectClasses
					.getClassesByObject(value);
			
			for (Class<?> clazz : classesList) {
				if (!kryoClasses.contains(clazz)) {
					try {
						if (storeKryoClasses(clazz))
							mark = 1;
					} catch (NumberFormatException e) {
						e.printStackTrace();
					} catch (Exception e) {
						// TODO Auto-generated catch block
						e.printStackTrace();
					}
				}
			}
		}
	}

	synchronized public static boolean storeKryoClasses(Class<?> clazz)
			throws NumberFormatException, Exception {
		if (getRemoteSize() > kryoMarkSize)
			updateLocalCache();
		storeElement(kryoClassesName, clazz.getName());
		int newCacheSize=kryoMarkSize+1;
		xmc.get("kryoClassesName" + kryoMarkSize + 1);
		if (xmc.add("kryoClassesName" + newCacheSize, kryoClassesName)) {
			 xmc.set(sizeMark, newCacheSize);
			storeElement(kryoClasses, clazz);
			xmc.delete("kryoClassesName" + kryoMarkSize);
			kryoMarkSize=newCacheSize;
			RegisterSerializers.registerClass(kryo, clazz);
			kryo.register(clazz);
			return true;
		}
		kryoClassesName.remove( clazz.getName());
		return false;
	}

	/**
	 * update local cache
	 * 
	 * @throws Exception
	 * @throws NumberFormatException
	 */
	@SuppressWarnings("unchecked")
	final static private void updateLocalCache() throws NumberFormatException,
			Exception {
		int markSize = getRemoteSize();
		synchronized (int.class) {
			ArrayList<String> kryoClassesName = (ArrayList<String>) xmc
					.get("kryoClasses" + markSize);
			for (int i = kryoClasses.size(); i < kryoClassesName.size(); i++) {
				Class<?> clazz = Class.forName(kryoClassesName.get(i), true,
						kryo.getClass().getClassLoader());
				kryo.register(clazz);
				kryoClasses.add(clazz);
			}
		}

	}

	final private static void storeElement(ArrayList v, Object value) {
		if (!v.contains(value)) {
			v.add(value);
		}
	}

	private static int getRemoteSize() throws Exception {
		Object markSize = xmc.get(sizeMark);
		kryoMarkSize = Integer.parseInt(markSize == null ? "0" : markSize + "");
		return kryoMarkSize;
	}

	/**
	 * deserialize the byteBuffer to Object
	 * 
	 * @param bytes
	 *            the array of bytes to be deserialize
	 * @return an object
	 * @throws Exception
	 */
	@SuppressWarnings("unchecked")
	final static public Object readObject(final byte[] bytes) throws Exception {
		Kryo kryo=new Kryo();
		registerInit(kryo);
		ByteBuffer buffer = ByteBuffer.allocateDirect(bytes.length);
		buffer.put(bytes);
		buffer.flip();
		Map<String, ByteArrayWarper> map = null;
		map = kryo.readObjectData(buffer, HashMap.class);
		Iterator<String> it = map.keySet().iterator();
		Object result = null;
		ByteBuffer bufferx = ByteBuffer.allocateDirect(1000);
		while (it.hasNext()) {
			String keyClassesName=it.next();
			String[] classesName = keyClassesName.split(" ");
			int length=classesName.length;
			ArrayList<Class<?>> classesList=new ArrayList<Class<?>>();
			for(int i=0;i<length;i++){
				if(!baseClassesName.contains(classesName[i]))
				classesList.add(Class.forName(classesName[i]));
			}
			ByteArrayWarper byteArrayWarrp = (ByteArrayWarper) map.get(keyClassesName);
			bufferx.put(byteArrayWarrp.getBytes());
			bufferx.flip();
			RegisterSerializers.registerAllClass(kryo, classesList);
			result = kryo.readObjectData(bufferx, classesList.get(0));
		}
		return result;
	}

	/**
	 * serialize the object to ByteBuffer the object first to be serialized to
	 * array of bytes ,the bytes are wraped to ByteArrayWarper the class of
	 * object as a key and the object of bytearraywarper as value be put in a
	 * map then serialize the map get the bytes ,this bytes will be return
	 * 
	 * @param o
	 *            the object to be serialized
	 * @return an array of bytes
	 */
	final static public byte[] writeObject(final Object o) {
		Kryo kryo=new Kryo();
		registerInit(kryo);
		List<Class<?>> classesList=ObjectClasses.getClassesByObject(o);
		StringBuffer classesName=new StringBuffer();
		for(Class<?> c:classesList){
			classesName.append(c.getName()+" ");
		}
		classesName.deleteCharAt(classesName.length()-1);
		RegisterSerializers.registerAllClass(kryo, ObjectClasses.getClassesByObject(o));
		ByteBuffer buffer = null;
		int size = 256;
		boolean isEnough = true;
		while (isEnough) {
			try {
				buffer = ByteBuffer.allocateDirect(size);
				// object serialize to ByteBuffer
				kryo.writeObjectData(buffer, o);
				buffer.flip();
				isEnough = false;
			} catch (BufferOverflowException ex) {
				size = 2 * size;
			}
		}
		final ByteArrayWarper byteArrayWarrp = new ByteArrayWarper();
		byte[] bytes = new byte[buffer.limit()];
		buffer.get(bytes);
		byteArrayWarrp.setBytes(bytes);
		final HashMap<String, ByteArrayWarper> classByte = new HashMap<String, ByteArrayWarper>();
		classByte.put(classesName.toString(), byteArrayWarrp);
		isEnough = true;
		ByteBuffer bufferx = ByteBuffer.allocateDirect(size);
		while (isEnough) {
			try {
				kryo.writeObjectData(bufferx, classByte);
				isEnough = false;
			} catch (BufferOverflowException ex) {
				size = 2 * size;
			}
		}
		bufferx.flip();
		byte[] result = new byte[bufferx.limit()];
		bufferx.get(result);
		return result;
	}

}
